package lib.paxos;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;

import lib.ArgumentValidator;
import lib.Command;
import lib.Utility;
import lib.logging.ILogger;
import lib.messaging.NetworkAddress;
import lib.rpc.IRPCMethods;

/**
 * Provides Paxos operations
 * @implements {@link IPaxosFunctionality}
 * @author abhip
 *
 */
public class PaxosFunctionality implements IPaxosFunctionality
{
    private class AcksAggregationResult
    {
        private List<AckMessage> acks;
        private Command largestAcceptedCommand;
        private int largestCounterValue;

        public AcksAggregationResult(
                int largestCounterValue,
                Command largestAcceptedCommand,
                List<AckMessage> acks)
        {
            ArgumentValidator.IsGreaterThan("largestCounterValue", largestCounterValue, 0);
            // largestAcceptedCommand can be null
            ArgumentValidator.IsNotNull("acks", acks);

            this.largestCounterValue = largestCounterValue;
            this.largestAcceptedCommand = largestAcceptedCommand;
            this.acks = acks;
        }

        public List<AckMessage> Acks()
        {
            return acks;
        }

        public Command LargestAcceptedCommand()
        {
            return largestAcceptedCommand;
        }

        public int LargestCounterValue()
        {
            return largestCounterValue;
        }

        @Override
        public String toString() {
            return "AcksAggregationResult [largestCounterValue="
                    + largestCounterValue + ", largestAcceptedCommand="
                    + largestAcceptedCommand + "]";
        }
    }

    private IAcceptorState acceptorState;
    private IRPCMethods clientRPCMethods;
    private ILogger logger;
    private Object result;
    private NetworkAddress serverAddress;
    private IRPCMethods serverRPCMethods;
    private boolean reconfigureAutomatically;

    public PaxosFunctionality(
            ILogger logger,
            NetworkAddress serverAddress,
            IRPCMethods serverRPCMethods,
            IRPCMethods clientRPCMethods,
            IAcceptorState acceptorState,
            boolean reconfigureAutomatically)
    {
        ArgumentValidator.IsNotNull("logger", logger);
        ArgumentValidator.IsNotNull("serverAddress", serverAddress);
        ArgumentValidator.IsNotNull("serverRPCMethods", serverRPCMethods);
        ArgumentValidator.IsNotNull("clientRPCMethods", clientRPCMethods);
        ArgumentValidator.IsNotNull("acceptorState", acceptorState);

        this.logger = logger;
        this.serverAddress = serverAddress;
        this.serverRPCMethods = serverRPCMethods;
        this.clientRPCMethods = clientRPCMethods;
        this.acceptorState = acceptorState;
        this.reconfigureAutomatically = reconfigureAutomatically;
    }

    private void BackoffExponentially(int retryCount)
    {
        double delta;

        if (retryCount <= 1)
        {
            delta = 0.0;
        }
        else if (retryCount <=2)
        {
            delta = 1.0;
        }
        else
        {
            delta = 2.0;
        }

        Random rn = new Random();
        double jitter = rn.nextDouble();

        try
        {
            logger.LogInfo(
                    "Going to sleep for %f sec as exponential backoff.",
                    delta + jitter);
            Thread.sleep((long) (delta + jitter) * 1000);
        }
        catch(InterruptedException e)
        {
            logger.LogException(e);
        }
    }

    private boolean DidEnoughServersReply(int replyCount, int sentMessagesCount)
    {
        if (sentMessagesCount == 1)
        {
            return (replyCount == sentMessagesCount);
        }

        int majorityCount = (sentMessagesCount + 1) / 2;
        int enoughServersCount = majorityCount + 1;
        if (replyCount >= enoughServersCount)
        {
            return true;
        }

        return false;
    }

    private boolean DidMajorityReply(int replyCount)
    {
        List<NetworkAddress> otherServerAddresses = acceptorState.GetOtherServerAddresses();

        int majorityCount = (otherServerAddresses.size() + 1) / 2;
        if (replyCount >= majorityCount)
        {
            return true;
        }

        return false;
    }

    private Object ExecuteCommandLocally(Command command)
    {
        try
        {
            if (command.CommandName().equalsIgnoreCase(IRPCMethods.CreateFileMethodName))
            {
                Object args[] = command.CommandArgs();
                int roundID = Integer.parseInt(args[0].toString());
                String fileName = (String) args[1];
                serverRPCMethods.CreateFile(this.serverAddress, this.serverAddress, roundID, fileName);
                return null;
            }

            if (command.CommandName().equalsIgnoreCase(IRPCMethods.AppendLineToFileMethodName))
            {
                Object args[] = command.CommandArgs();
                int roundID = Integer.parseInt(args[0].toString());
                String fileName = (String) args[1];
                String data = (String) args[2];
                serverRPCMethods.AppendLineToFile(this.serverAddress, this.serverAddress, roundID, fileName, data);
                return null;
            }

            if (command.CommandName().equalsIgnoreCase(IRPCMethods.GetLineFromFileMethodName))
            {
                Object args[] = command.CommandArgs();
                int roundID = Integer.parseInt(args[0].toString());
                String fileName = (String) args[1];
                int lineNumber = (Integer) args[2];
                return serverRPCMethods.GetLineFromFile(
                        this.serverAddress,
                        this.serverAddress,
                        roundID,
                        fileName,
                        lineNumber);
            }

            if (command.CommandName().equalsIgnoreCase(IRPCMethods.JoinMethodName))
            {
                Object args[] = command.CommandArgs();
                int roundID = Integer.parseInt(args[0].toString());
                NetworkAddress newServerAddress = (NetworkAddress) args[1];
                return serverRPCMethods.Join(this.serverAddress, this.serverAddress, roundID, newServerAddress);
            }

            if (command.CommandName().equalsIgnoreCase(IRPCMethods.ReconfigureMethodName))
            {
                Object args[] = command.CommandArgs();
                int roundID = Integer.parseInt(args[0].toString());

                @SuppressWarnings("unchecked")
                List<NetworkAddress> newServers = (List<NetworkAddress>) args[1];

                serverRPCMethods.Reconfigure(this.serverAddress, this.serverAddress, roundID, newServers);
                return null;
            }

            logger.LogError(
                    "Unknown command tried to be executed locally. command:%s",
                    command.toString());
            return null;
        }
        catch (IOException e)
        {
            logger.LogException(e);
            return e;
        }
        catch (ClassNotFoundException e)
        {
            logger.LogException(e);
            return e;
        }
    }

    private AcksAggregationResult GetAggregateResultFromAcks(
            List<AckMessage> ackMessages,
            int counterValue) throws PaxosException
    {
        int acceptorRoundID;
        SequenceID highestAcceptorSeqID = null;
        AcceptMessage highestAcceptedMessage = null;

        for (int i = 0; i < ackMessages.size(); ++i)
        {
            AckMessage ack = ackMessages.get(i);

            acceptorRoundID = ack.CurrentRoundID();
            if (acceptorRoundID > this.acceptorState.CurrentRoundID())
            {
                throw new PaxosException(
                        String.format(
                                "Possibly failed updating server state during some paxos phase. " +
                                        "acceptorRoundIDs:%d should have be less than or equal to serverRoundID:%d",
                                acceptorRoundID,
                                this.acceptorState.CurrentRoundID()));
            }

            if (acceptorRoundID < this.acceptorState.CurrentRoundID())
            {
                logger.LogInfo(
                        "Acceptor is probably out of date wrt paxos rounds, ignoring it's ack. " +
                            "acceptor:%s, acceptor's latest roundID:%d, proposer's currentRoundID:%d.",
                        ack.SenderAddress(),
                        ack.CurrentRoundID(),
                        this.acceptorState.CurrentRoundID());
                continue;
            }

            if (ack.HighestKnownSeqID() != null)
            {
                if (highestAcceptorSeqID == null ||
                        highestAcceptorSeqID.IsLessThan(ack.HighestKnownSeqID()))
                {
                    highestAcceptorSeqID = ack.HighestKnownSeqID();
                }
            }

            if (ack.AcceptedMessage() != null)
            {
                if (highestAcceptedMessage == null ||
                        highestAcceptedMessage.SequenceID().IsLessThan(ack.AcceptedMessage().SequenceID()))
                {
                    highestAcceptedMessage = ack.AcceptedMessage();
                }
            }
        }

        int highestCounterValue = (highestAcceptorSeqID == null) ?
                counterValue : highestAcceptorSeqID.Counter();
        Command largestAcceptedCommand = (highestAcceptedMessage == null) ?
                null : (highestAcceptedMessage.Command());

        return new AcksAggregationResult(highestCounterValue, largestAcceptedCommand, ackMessages);
    }

    private List<NetworkAddress> GetListOfOnlineServers(List<AckMessage> acks)
    {
        List<NetworkAddress> onlineServers = new ArrayList<NetworkAddress>();

        for (int i = 0; i < acks.size(); ++i)
        {
            onlineServers.add(acks.get(i).SenderAddress());
        }

        onlineServers.add(this.serverAddress);

        return onlineServers;
    }

    private List<RoundInfo> GetMissingRoundInfos(int acceptorRoundID)
    {
        ArgumentValidator.IsBetweenBothInclusive(
                "acceptorRoundID",
                acceptorRoundID,
                0,
                acceptorState.GetTotalNumberOfCommands());

        List<RoundInfo> missingRoundInfos = new ArrayList<RoundInfo>();
        for (int i = acceptorRoundID + 1; i <= acceptorState.GetTotalNumberOfCommands(); ++i)
        {
            try
            {
                missingRoundInfos.add(acceptorState.GetRoundInfo(i));
            }
            catch (PaxosException e)
            {
                logger.LogException(e);
                continue;
            }
        }

        return missingRoundInfos;
    }

    @Override
    public Object GetResult()
    {
        return result;
    }

    private boolean IsCurrentServerPartOfEpochBasedOnAcks(List<AckMessage> ackMessages)
    {
        int acceptorRoundID;

        for (int i = 0; i < ackMessages.size(); ++i)
        {
            AckMessage ack = ackMessages.get(i);

            acceptorRoundID = ack.CurrentRoundID();
            if (acceptorRoundID < (this.acceptorState.CurrentRoundID() - 1))
            {
                logger.LogInfo(
                        "Acceptor is probably out of date wrt paxos rounds, " +
                            "ignoring it's ack for seeing if current server is part of epoch. " +
                            "acceptor:%s, acceptor's latest roundID:%d, proposer's currentRoundID:%d.",
                        ack.SenderAddress(),
                        ack.CurrentRoundID(),
                        this.acceptorState.CurrentRoundID());
                continue;
            }

            if (!ack.IsProposerPartOfEpoch())
            {
                return false;
            }
        }

        return true;
    }

    private AcksAggregationResult RunAcceptPhase(
            int roundID,
            int counterValue,
            Command commandToAccept,
            int retryCount) throws PaxosException
    {
        SequenceID seqID = new SequenceID(serverAddress.Port(), counterValue);
        AcceptMessage acceptMsg = new AcceptMessage(roundID, seqID, commandToAccept);

        if (retryCount == 0)
        {
            acceptorState.UpdateAcceptorState(
                    roundID,
                    seqID,
                    acceptMsg,
                    new ArrayList<RoundInfo>());
        }

        List<AckMessage> acks = SendMessageToOtherServers(acceptMsg);
        List<NetworkAddress> otherServerAddresses = acceptorState.GetOtherServerAddresses();

        int ackCount = acks.size();
        if(!DidMajorityReply(ackCount))
        {
            // Keep retrying till majority replies!
            logger.LogError(
                    "Majority did not reply during decide phase! " +
                        "Retrying after exponential backoff. " +
                        "RetryCount:%d, RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                    retryCount + 1,
                    acceptorState.CurrentRoundID(),
                    acks.size(),
                    otherServerAddresses.size(),
                    Utility.FlattenListToString(acks));
            BackoffExponentially(retryCount);
            return RunAcceptPhase(roundID, counterValue, commandToAccept, ++retryCount);
        }

        logger.LogInfo(
                "Wohoo! Majority replied to accept request!!" +
                    "RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                acceptorState.CurrentRoundID(),
                ackCount,
                otherServerAddresses.size(),
                Utility.FlattenListToString(acks));

        if (!IsCurrentServerPartOfEpochBasedOnAcks(acks))
        {
            logger.LogError(
                    "Other servers during accept phase said current server (%s) is no longer part of epoch. " +
                        "Please kill this server and restart it in join mode. Other potential server:%s",
                    this.serverAddress,
                    acks.get(0).SenderAddress());

            throw new PaxosException(
                    String.format(
                            "Server %s is no longer part of epoch. Please try some other server. Other potential server:%s",
                            this.serverAddress,
                            acks.get(0).SenderAddress()));
        }

        UpdateAndSyncBasedOnAcksIfRequired(acks);

        logger.LogInfo("Update and sync complete. Going to aggregate accept acks.");
        AcksAggregationResult acksAggResult = GetAggregateResultFromAcks(acks, counterValue);
        if (acksAggResult.LargestAcceptedCommand() == null)
        {
            throw new PaxosException(
                    String.format(
                            "During accept phase, none of acceptors accepted the proposed command command! " +
                                "RoundID:%d, Servers replied:%s",
                            acceptorState.CurrentRoundID(),
                            Utility.FlattenListToString(acks)));
        }

        logger.LogInfo("Aggregation of accept acks complete! %s", acksAggResult);
        return acksAggResult;
    }

    private AcksAggregationResult RunDecidePhase(
            int roundID,
            int counterValue,
            Command commandToDecide,
            int retryCount) throws PaxosException
    {
        SequenceID seqID = new SequenceID(serverAddress.Port(), counterValue);
        DecideMessage decideMsg = new DecideMessage(roundID, seqID, commandToDecide);
        List<AckMessage> acks = SendMessageToOtherServers(decideMsg);

        if (retryCount == 0)
        {
            List<RoundInfo> roundInfos = new ArrayList<RoundInfo>();
            roundInfos.add(new RoundInfo(roundID, decideMsg.Command()));
            acceptorState.UpdateAcceptorState(
                    roundID,
                    seqID,
                    acceptorState.AcceptedMessage(),
                    roundInfos);
        }

        int ackCount = acks.size();
        List<NetworkAddress> otherServerAddresses = acceptorState.GetOtherServerAddresses();
        if(!DidMajorityReply(ackCount))
        {
            // Keep retrying till majority replies!
            logger.LogError(
                    "Majority did not reply during decide phase! " +
                        "Retrying after exponential backoff. " +
                        "RetryCount:%d, RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                    retryCount + 1,
                    acceptorState.CurrentRoundID(),
                    acks.size(),
                    otherServerAddresses.size(),
                    Utility.FlattenListToString(acks));
            BackoffExponentially(retryCount);
            return RunDecidePhase(roundID, counterValue, commandToDecide, retryCount + 1);
        }

        logger.LogInfo(
                "Wohoo! Majority replied to decide request!!" +
                    "RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                acceptorState.CurrentRoundID(),
                ackCount,
                otherServerAddresses.size(),
                Utility.FlattenListToString(acks));

        if (!IsCurrentServerPartOfEpochBasedOnAcks(acks))
        {
            logger.LogError(
                    "Other servers during decide phase said current server (%s) is no longer part of epoch. " +
                        "Please kill this server and restart it in join mode. Other potential server:%s",
                    this.serverAddress,
                    acks.get(0).SenderAddress());

            throw new PaxosException(
                    String.format(
                            "Server %s is no longer part of epoch. Please try some other server. Other potential server:%s",
                            this.serverAddress,
                            acks.get(0).SenderAddress()));
        }

        UpdateAndSyncBasedOnAcksIfRequired(acks);

        logger.LogInfo("Update and sync complete. Going to aggregate decide acks.");

        AcksAggregationResult acksAggResult = GetAggregateResultFromAcks(acks, counterValue);
        if (acksAggResult.LargestAcceptedCommand() == null)
        {
            throw new PaxosException(
                    String.format(
                            "During decide phase, none of acceptors accepted the proosed command! " +
                                "RoundID:%d, Servers replied:%s",
                            acceptorState.CurrentRoundID(),
                            Utility.FlattenListToString(acks)));
        }

        logger.LogInfo("Aggregation of decide acks complete! %s", acksAggResult);
        return acksAggResult;
    }

    @Override
    public boolean RunPaxos(
            int roundID,
            Command clientCommand)
    {
       return RunPaxos(roundID, 1, clientCommand, 0);
    }

    private boolean RunPaxos(
            int roundID,
            int counterValue,
            Command command,
            int retryCount)
    {
        try
        {
            VerifyRoundID(roundID);

            Command commandToPropose;

            if (this.acceptorState.NextEpochServerAddresses() == null)
            {
                commandToPropose = command;
            }
            else
            {
                logger.LogError(
                        "Cannot propose client command '%s' because reconfiguration is needed. " +
                            "Trying to run Paxos to gain consensus on next epoch:%s",
                        command,
                        Utility.FlattenListToString(this.acceptorState.NextEpochServerAddresses()));

                commandToPropose = new Command(
                        IRPCMethods.ReconfigureMethodName,
                        new Serializable[] { roundID, (Serializable) this.acceptorState.NextEpochServerAddresses() });
            }

            AcksAggregationResult prepareAcksAggResult = RunPreparePhase(roundID, counterValue, 0);
            Command commandToAccept = (prepareAcksAggResult.LargestAcceptedCommand() == null) ?
                    commandToPropose : prepareAcksAggResult.LargestAcceptedCommand();

            if (prepareAcksAggResult.LargestCounterValue() > counterValue)
            {
                // re-run Paxos with higher counterValue and commandToAccept
                BackoffExponentially(retryCount);
                logger.LogError(
                        "Acceptor reported higher sequenceID during prepare phase! " +
                            "Retrying after exponential backoff. " +
                            "RoundID:%d, Acceptor's counters  %s, local counter: %s, RetryCount:%d",
                        roundID,
                        prepareAcksAggResult.LargestCounterValue(),
                        counterValue,
                        retryCount + 1);

                return RunPaxos(
                        roundID,
                        prepareAcksAggResult.LargestCounterValue() + 1,
                        commandToAccept,
                        retryCount + 1);
            }

            logger.LogInfo(
                    "Prepare phase completed successfully, starting accept phase. " +
                            "RoundID: %d, Command:%s",
                    roundID,
                    commandToAccept.toString());

            AcksAggregationResult acceptAcksAggResult = RunAcceptPhase(
                    roundID,
                    counterValue,
                    commandToAccept,
                    0);
            Command commandToDecide = acceptAcksAggResult.LargestAcceptedCommand();

            if (acceptAcksAggResult.LargestCounterValue() > counterValue)
            {
                // re-run Paxos with higher counterValue and commandToDecide
                BackoffExponentially(retryCount);
                logger.LogError(
                        "Acceptor reported higher sequenceID during accept phase! " +
                            "Retrying paxos after exponential backoff. " +
                            "RoundID:%d, Acceptor's counters  %s, local counter: %s, RetryCount:%d",
                        roundID,
                        acceptAcksAggResult.LargestCounterValue(),
                        counterValue,
                        retryCount + 1);

                return RunPaxos(
                        roundID,
                        acceptAcksAggResult.LargestCounterValue() + 1,
                        commandToDecide,
                        retryCount + 1);
            }

            logger.LogInfo(
                    "Accept phase completed successfully, starting decide phase. RoundID: %d, Command:%s",
                    roundID,
                    commandToDecide.toString());

            AcksAggregationResult decideAcksAggResult = RunDecidePhase(
                    roundID,
                    counterValue,
                    commandToDecide,
                    0);
            Command decidedCommand = decideAcksAggResult.LargestAcceptedCommand();

            if (decideAcksAggResult.LargestCounterValue() > counterValue)
            {
                // re-run Paxos with higher counterValue and commandToDecide
                BackoffExponentially(retryCount);
                logger.LogError(
                        "Acceptor reported higher sequenceID during decide phase! " +
                            "Retrying paxos after exponential backoff. " +
                            "RoundID:%d, Acceptor's counters  %s, local counter: %s, RetryCount:%d",
                        roundID,
                        decideAcksAggResult.LargestCounterValue(),
                        counterValue,
                        retryCount + 1);

                return RunPaxos(
                        roundID,
                        decideAcksAggResult.LargestCounterValue() + 1,
                        decidedCommand,
                        retryCount + 1);
            }

            logger.LogInfo(
                    "Decide phase completed successfully, going to execute locally. RoundID: %d, Command:%s",
                    roundID,
                    decidedCommand.toString());

            this.result = ExecuteCommandLocally(decidedCommand);

            logger.LogInfo("Command successfully executed locally! " +
                                "roundID:%d, command:%s, result:%s",
                            roundID,
                            decidedCommand.toString(),
                            this.result);

            int sentMessagesCount = this.acceptorState.GetOtherServerAddresses().size();

            if (decidedCommand.CommandName().equals(IRPCMethods.JoinMethodName))
            {
                // if it was a join command, then new server has been just added to the state,
                // but it did not participate in this paxos round,
                // so sentMessagesCount is 1 less than the number of Paxos servers
                --sentMessagesCount;
            }

            if (!DidEnoughServersReply(decideAcksAggResult.Acks().size(), sentMessagesCount))
            {
                if (this.reconfigureAutomatically)
                {
                    List<NetworkAddress> onlineServers = GetListOfOnlineServers(decideAcksAggResult.Acks());
                    Collections.sort(onlineServers);
                    this.acceptorState.SetNextEpochServerAddresses(onlineServers);

                    logger.LogError(
                            "The number of alive servers has become critically low. " +
                                "Servers alive:%d/%d. Will try to reconfigure system soon. Potential next epoch:%s",
                            decideAcksAggResult.Acks().size() + 1,
                            this.acceptorState.GetOtherServerAddresses().size() + 1,
                            Utility.Sort(onlineServers));
                }
                else
                {
                    logger.LogError(
                            "The number of alive servers has become critically low, but automatic reconfiguration is disabled. " +
                                "Servers alive:%d/%d.",
                            decideAcksAggResult.Acks().size() + 1,
                            this.acceptorState.GetOtherServerAddresses().size() + 1);

                    this.acceptorState.SetNextEpochServerAddresses(null);
                }
            }
            else
            {
                if (this.acceptorState.GetOtherServerAddresses().size() == 1)
                {
                    logger.LogError(
                            "Warning! System is operating with minimum number of servers i.e. %d. " +
                                "Please consider adding more servers...",
                                this.acceptorState.GetOtherServerAddresses().size() + 1);
                }
                else
                {
                    logger.LogInfo(
                        "Sufficient number of servers are alive. No need for reconfiguration. Servers alive:%d/%d.",
                        decideAcksAggResult.Acks().size() + 1,
                        this.acceptorState.GetOtherServerAddresses().size() + 1);
                }

                this.acceptorState.SetNextEpochServerAddresses(null);
            }

            return true;
        }
        catch (PaxosException e)
        {
            this.result = e;
            return false;
        }
    }

    private AcksAggregationResult RunPreparePhase(
            int roundID,
            int counterValue,
            int retryCount) throws PaxosException
    {
        SequenceID seqID = new SequenceID(serverAddress.Port(), counterValue);
        PrepareMessage prepareMsg = new PrepareMessage(roundID, seqID);

        if (retryCount == 0)
        {
            acceptorState.UpdateAcceptorState(
                    roundID,
                    seqID,
                    null,
                    new ArrayList<RoundInfo>());
        }

        List<AckMessage> acks = SendMessageToOtherServers(prepareMsg);

        List<NetworkAddress> otherServerAddresses = acceptorState.GetOtherServerAddresses();
        int ackCount = acks.size();
        if(!DidMajorityReply(ackCount))
        {
            // Keep retrying till majority replies!
            logger.LogError(
                    "Majority did not reply during prepare phase! " +
                        "Retrying after exponential backoff. " +
                        "RetryCount:%d, RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                    retryCount + 1,
                    acceptorState.CurrentRoundID(),
                    acks.size(),
                    otherServerAddresses.size(),
                    Utility.FlattenListToString(acks));

            BackoffExponentially(retryCount);
            return RunPreparePhase(roundID, counterValue, ++retryCount);
        }

        logger.LogInfo(
                "Wohoo! Majority replied to prepare request!!" +
                    "RoundID:%d, Reply count:%d/%d. Servers replied:%s",
                acceptorState.CurrentRoundID(),
                ackCount,
                otherServerAddresses.size(),
                Utility.FlattenListToString(acks));

        if (!IsCurrentServerPartOfEpochBasedOnAcks(acks))
        {
            logger.LogError(
                    "Other servers during prepare phase said current server (%s) is no longer part of epoch. " +
                        "Please kill this server and restart it in join mode. Other potential server:%s",
                    this.serverAddress,
                    acks.get(0).SenderAddress());

            throw new PaxosException(
                    String.format(
                            "Server %s is no longer part of epoch. Please try some other server. Other potential server:%s",
                            this.serverAddress,
                            acks.get(0).SenderAddress()));
        }

        UpdateAndSyncBasedOnAcksIfRequired(acks);

        logger.LogInfo("Update and sync complete. Going to aggregate prepare acks.");

        AcksAggregationResult acksAggResult = GetAggregateResultFromAcks(acks, counterValue);
        logger.LogInfo("Aggregation of prepare acks complete! %s", acksAggResult);
        return acksAggResult;
    }

    private List<AckMessage> SendMessageToOtherServers(Object message)
    {
        List<AckMessage> ackMessages = new ArrayList<AckMessage>();
        List<NetworkAddress> otherServerAddresses = acceptorState.GetOtherServerAddresses();

        for (int i = 0; i < otherServerAddresses.size(); ++i)
        {
            try
            {
                if (message instanceof PrepareMessage)
                {
                    ackMessages.add(clientRPCMethods.Prepare(
                            serverAddress,
                            otherServerAddresses.get(i),
                            (PrepareMessage) message));
                }
                else if (message instanceof AcceptMessage)
                {
                    ackMessages.add(clientRPCMethods.Accept(
                            serverAddress,
                            otherServerAddresses.get(i),
                            (AcceptMessage) message));
                }
                else if (message instanceof DecideMessage)
                {
                    ackMessages.add(clientRPCMethods.Decide(
                            serverAddress,
                            otherServerAddresses.get(i),
                            (DecideMessage) message));
                }
                else if (message instanceof SyncMessage)
                {
                    ackMessages.add(clientRPCMethods.Sync(
                            serverAddress,
                            otherServerAddresses.get(i),
                            (SyncMessage) message));
                }
                else
                {
                    throw new IllegalArgumentException(String.format(
                            "Cannot send message of type: %s",
                            message.getClass()));
                }
            }
            catch (IOException e)
            {
                logger.LogException(e);
                continue;
            }
            catch (PaxosException e)
            {
                logger.LogException(e);
                continue;
            }
            catch (ClassNotFoundException e)
            {
                logger.LogException(e);
                continue;
            }
            catch (ClassCastException e)
            {
                logger.LogException(e);
                continue;
            }
        }

        return ackMessages;
    }

    private void SendSyncMessagesIfRequired(List<AckMessage> acks) throws PaxosException
    {
        for (int i = 0; i < acks.size(); ++i)
        {
            AckMessage ack = acks.get(i);

            if (ack.CurrentRoundID() >= acceptorState.CurrentRoundID())
            {
                continue;
            }

            if ((ack.CurrentRoundID() == acceptorState.CurrentRoundID() - 1) &&
                    !acceptorState.IsCommandChosenForCurrentRound())
            {
                continue;
            }

            try
            {
                logger.LogError(
                        "Acceptor %s is out of date. Sending sync mesaage. " +
                            "Current roundID:%d, Acceptor roundID:%d",
                        ack.SenderAddress(),
                        this.acceptorState.CurrentRoundID(),
                        ack.CurrentRoundID());

                List<RoundInfo> missingRoundInfos = GetMissingRoundInfos(ack.CurrentRoundID());
                SyncMessage syncMsg = new SyncMessage(
                        acceptorState.CurrentRoundID(),
                        acceptorState.HighestKnownSeqID(),
                        missingRoundInfos);

                logger.LogError(
                        "Sync messge details: %s",
                        syncMsg.toString());

                this.clientRPCMethods.Sync(this.serverAddress, ack.SenderAddress(), syncMsg);
            }
            catch (ClassNotFoundException e)
            {
                logger.LogException(e);
                continue;
            }
            catch (IOException e)
            {
                logger.LogException(e);
                continue;
            }
            catch (PaxosException e)
            {
                logger.LogException(e);
                continue;
            }
        }
    }

    private void UpdateAndSyncBasedOnAcksIfRequired(List<AckMessage> ackMessages) throws PaxosException
    {
        boolean wasServerStateUpdated = false;

        if ((ackMessages != null) && (ackMessages.size() > 0))
        {
            wasServerStateUpdated = UpdateStateIfRequired(ackMessages);
            SendSyncMessagesIfRequired(ackMessages);
        }

        if (wasServerStateUpdated)
        {
            throw new PaxosException(
                    String.format(
                            "Server %s updated with missing Paxos rounds. This server now knows about %d rounds. " +
                                "Please retry with higher roundID.",
                            this.serverAddress,
                            this.acceptorState.CurrentRoundID()));
        }
    }

    private boolean UpdateStateIfRequired(List<AckMessage> ackMessages) throws PaxosException
    {
        int maxRoundIDIndex = 0;
        int maxRoundID = ackMessages.get(0).CurrentRoundID();

        for (int i = 1; i < ackMessages.size(); ++i)
        {
            if (maxRoundID < ackMessages.get(i).CurrentRoundID())
            {
                maxRoundIDIndex = i;
                maxRoundID = ackMessages.get(i).CurrentRoundID();
            }
        }

        if (maxRoundID <= acceptorState.CurrentRoundID())
        {
            logger.LogInfo("Current server is up to date wrt paxos rounds. server:%s, currentRoundID:%d",
                    serverAddress.toString(),
                    this.acceptorState.CurrentRoundID());
            return false;
        }

        AckMessage ackMsg = ackMessages.get(maxRoundIDIndex);

        logger.LogError(
                "Current server is out of date wrt paxos rounds! " +
                    "Current server:%s, Current currentRoundID:%d, " +
                        "server with latest info:%s, latest roundID:%d, missing number of commands:%d",
                serverAddress.toString(),
                this.acceptorState.CurrentRoundID(),
                ackMsg.SenderAddress().toString(),
                maxRoundID,
                maxRoundID - acceptorState.CurrentRoundID());

        List<RoundInfo> missingRoundInfos = ackMessages.get(maxRoundIDIndex).MissingRoundInfos();
        for (int i = 0; i < missingRoundInfos.size(); ++i)
        {
            logger.LogInfo("Applying missing command: %s", missingRoundInfos.get(i));
            ExecuteCommandLocally(missingRoundInfos.get(i).Command());
        }

        acceptorState.UpdateAcceptorState(
                ackMsg.CurrentRoundID(),
                ackMsg.HighestKnownSeqID(),
                ackMsg.AcceptedMessage(),
                ackMsg.MissingRoundInfos());

        return true;
    }

    private void VerifyRoundID(int roundID) throws PaxosException
    {
        if (roundID < acceptorState.CurrentRoundID())
        {
            throw new PaxosException(
                    String.format(
                            "For roundID %d, already chose command '%s'. Next available roundID is '%d'.",
                            roundID,
                            acceptorState.GetRoundInfo(roundID).toString(),
                            acceptorState.CurrentRoundID() + 1));
        }

        if (roundID == acceptorState.CurrentRoundID())
        {
            if (acceptorState.IsCommandChosenForCurrentRound())
            {
                throw new PaxosException(
                        String.format(
                                "For roundID %d, already chose command '%s'. Next available roundID is '%d'.",
                                roundID,
                                acceptorState.GetRoundInfo(roundID).toString(),
                                acceptorState.CurrentRoundID() + 1));
            }

            logger.LogInfo(
                    "Given roundID equals currentRoundID, " +
                        "but no command has been chosen yet for currentRoundID. server:%s, currentRoundID:%d",
                    this.serverAddress,
                    acceptorState.CurrentRoundID());

            // valid roundID, since no command has be chosen for this round
            return;
        }

        if (roundID == acceptorState.CurrentRoundID() + 1)
        {
            // valid roundID
            return;
        }

        throw new PaxosException(
                String.format(
                        "RoundID %d is too large. Please use '%d' instead.",
                        roundID,
                        acceptorState.CurrentRoundID() + 1));
    }
}
